﻿using System;
using System.Collections.Generic;
using __ts.Core.Api;
using __ts.Core.FilterChain;
using __ts.Core.Service;
using __ts.Core.Util.IO;
using log4net;

/*
 *  author: able
 *  @2012/2/21 14:39:12
 * 
 */
namespace __ts.Core.Session
{
    /// <summary>
    /// 
    /// </summary>
    public abstract class AbstractIoSession : IoSession, ReadFilterChainController, WriteFilterChainController
    {
        /** The logger for this class */
    private static ILog LOG = LogManager.GetLogger(typeof(AbstractIoSession));

    /** unique identifier generator */
    //private static AtomicInteger NEXT_ID = new AtomicInteger(0);

    /** The session's unique identifier */
    private long id;

    /** The session's creation time */
    private long creationTime;

    /** The service this session is associated with */
    private IoService service;

    /** attributes map */
    //private AttributeContainer attributes = new DefaultAttributeContainer();

    /** The {@link SelectorProcessor} used for handling this session writing */
    protected SelectorProcessor writeProcessor;

    //------------------------------------------------------------------------
    // Basic statistics
    //------------------------------------------------------------------------

    /** The number of bytes read since this session has been created */
    private long readBytes;

    /** The number of bytes written since this session has been created */
    private long writtenBytes;

    /** Last time something was read for this session */
    private long lastReadTime;

    /** Last time something was written for this session */
    private long lastWriteTime;

    //------------------------------------------------------------------------
    // Session state
    //------------------------------------------------------------------------

    /** The session's state : one of CREATED, CONNECTED, CLOSING, CLOSED, SECURING, CONNECTED_SECURED */
    protected volatile SessionState state;

    /** A lock to protect the access to the session's state */
    //private ReadWriteLock stateLock = new ReentrantReadWriteLock();

    ///** A Read lock on the reentrant session's state lock */
    //private Lock stateReadLock = stateLock.readLock();

    ///** A Write lock on the reentrant session's state lock */
    //private Lock stateWriteLock = stateLock.writeLock();

    ///** Tells if the session is secured or not */
    protected volatile bool secured;

    ///** is this session registered for being polled for write ready events */
    //private AtomicBoolean registeredForWrite = new AtomicBoolean();

    ////------------------------------------------------------------------------
    //// Write queue
    ////------------------------------------------------------------------------

    ///** the queue of pending writes for the session, to be dequeued by the {@link SelectorProcessor} */
    //private Queue<WriteRequest> writeQueue = new DefaultWriteQueue();
    private Queue<WriteRequest> writeQueue = new Queue<WriteRequest>();

    ///** A lock to protect the access to the write queue */
    //private ReadWriteLock writeQueueLock = new ReentrantReadWriteLock();

    ///** A Read lock on the reentrant writeQueue lock */
    //private Lock writeQueueReadLock = writeQueueLock.readLock();

    ///** A Write lock on the reentrant writeQueue lock */
    //private Lock writeQueueWriteLock = writeQueueLock.writeLock();

    //------------------------------------------------------------------------
    // Filter chain
    //------------------------------------------------------------------------

    /** The list of {@link IoFilter} implementing this chain. */
    private IoFilter[] chain;

    /** the current position in the write chain for this thread */
    private int writeChainPosition;

    /** the current position in the read chain for this thread */
    private int readChainPosition;

    /** hold the last WriteRequest created for the high level message currently written (can be null) */
    private WriteRequest lastWriteRequest;

    /**
     * Create an {@link org.apache.mina.api.IoSession} with a unique identifier (
     * {@link org.apache.mina.api.IoSession#getId()}) and an associated {@link IoService}
     * 
     * @param service the service this session is associated with
     * @param writeProcessor the processor in charge of processing this session write queue
     */
    public AbstractIoSession(IoService service, SelectorProcessor writeProcessor) {
        // generated a unique id
        //id = NEXT_ID.getAndIncrement();
        //creationTime = System.currentTimeMillis();
        //this.service = service;
        //this.writeProcessor = writeProcessor;
        //this.chain = service.getFilters();

        //LOG.debug("Created new session with id : {}", id);

        //this.state = SessionState.CREATED;
    }

    public bool isClosed() {
        try {
            //stateReadLock.lock();

            return state == SessionState.CLOSED;
        } finally {
            //stateReadLock.unlock();
        }
    }

    /**
     * {@inheritDoc}
     */
    public bool isClosing() {
        try {
            //stateReadLock.lock();

            return state == SessionState.CLOSING;
        } finally {
            //stateReadLock.unlock();
        }
    }

    /**
     * {@inheritDoc}
     */
    public bool isConnected() {
        try {
            //stateReadLock.lock();

            return state == SessionState.CONNECTED;
        } finally {
            //stateReadLock.unlock();
        }
    }

    /**
     * {@inheritDoc}
     */
    public bool isCreated() {
        try {
            //stateReadLock.lock();

            return state == SessionState.CREATED;
        } finally {
            //stateReadLock.unlock();
        }
    }

    /**
     * {@inheritDoc}
     */
    public bool isSecuring() {
        try {
            //stateReadLock.lock();

            return state == SessionState.SECURING;
        } finally {
            //stateReadLock.unlock();
        }
    }

    /**
     * {@inheritDoc}
     */
    public bool isConnectedSecured() {
        try {
            //stateReadLock.lock();

            return state == SessionState.SECURED;
        } finally {
            //stateReadLock.unlock();
        }
    }

    /**
     * {@inheritDoc}
     */
    public void changeState(SessionState to) {
        //try {
        //    //stateWriteLock.lock();

        //    switch (state) {
        //    case CREATED:
        //        switch (to) {
        //        case CONNECTED:
        //        case SECURING:
        //        case CLOSING:
        //            state = to;
        //            break;

        //        default:
        //            throw new IllegalStateException("Cannot transit from " + state + " to " + to);
        //        }

        //        break;

        //    case CONNECTED:
        //        switch (to) {
        //        case SECURING:
        //        case CLOSING:
        //            state = to;
        //            break;

        //        default:
        //            throw new IllegalStateException("Cannot transit from " + state + " to " + to);
        //        }

        //        break;

        //    case SECURING:
        //        switch (to) {
        //        case SECURED:
        //        case CLOSING:
        //            state = to;
        //            break;

        //        default:
        //            throw new IllegalStateException("Cannot transit from " + state + " to " + to);
        //        }

        //        break;

        //    case SECURED:
        //        switch (to) {
        //        case CONNECTED:
        //        case SECURING:
        //        case CLOSING:
        //            state = to;
        //            break;

        //        default:
        //            throw new IllegalStateException("Cannot transit from " + state + " to " + to);
        //        }

        //        break;
        //    case CLOSING:
        //        if (to != SessionState.CLOSED) {
        //            throw new IllegalStateException("Cannot transit from " + state + " to " + to);
        //        }

        //        state = to;

        //        break;

        //    case CLOSED:
        //        throw new IllegalStateException("The session is already closed. cannot switch to " + to);
        //    }
        //} finally {
        //    stateWriteLock.unlock();
        //}
    }

    //------------------------------------------------------------------------
    // SSL/TLS session state management
    //------------------------------------------------------------------------
    /**
     * {@inheritDoc}
     */
    public bool isSecured() {
        return secured;
    }

    /**
     * {@inheritDoc}
     */
    public void setSecured(bool secured) {
        this.secured = secured;
    }

    /**
     * {@inheritDoc}
     */
    //public void initSecure(SSLContext sslContext){
        //SslHelper sslHelper = new SslHelper(this, sslContext);
        //sslHelper.init();

        //attributes.setAttribute(SSL_HELPER, sslHelper);
        //setSecured(true);
    //}

    /**
     * {@inheritDoc}
     */
    public long getId() {
        return id;
    }

    /**
     * {@inheritDoc}
     */
    public long getCreationTime() {
        return creationTime;
    }

    /**
     * {@inheritDoc}
     */
    public long getReadBytes() {
        return readBytes;
    }

    /**
     * To be called by the internal plumber when some bytes are written on the socket
     * @param bytesCount number of extra bytes written
     */
    public void incrementWrittenBytes(int bytesCount) {
        writtenBytes += bytesCount;
    }

    /**
     * {@inheritDoc}
     */
    public long getWrittenBytes() {
        return writtenBytes;
    }

    /**
     * {@inheritDoc}
     */
    public long getLastReadTime() {
        return lastReadTime;
    }

    /**
     * {@inheritDoc}
     */
    public long getLastWriteTime() {
        return lastWriteTime;
    }

    /**
     * {@inheritDoc}
     */
    public long getLastIoTime() {
        return Math.Max(lastReadTime, lastWriteTime);
    }

    /**
     * {@inheritDoc}
     */
    public IoService getService() {
        return service;
    }

    /**
     * {@inheritDoc}
     * 
     * @exception IllegalArgumentException if <code>key==null</code>
     * @see #setAttribute(AttributeKey, Object)
     */
    //public <T> T getAttribute(AttributeKey<T> key, T defaultValue) {
    //    return attributes.getAttribute(key, defaultValue);
    //}

    /**
     * {@inheritDoc}
     * 
     * @exception IllegalArgumentException if <code>key==null</code>
     * @see #setAttribute(AttributeKey, Object)
     */
    //@Override
    //public <T> T getAttribute(AttributeKey<T> key) {
    //    return attributes.getAttribute(key);
    //}

    /**
     * {@inheritDoc}
     * 
     * @exception IllegalArgumentException
     * <ul>
     *   <li>
     *     if <code>key==null</code>
     *   </li>
     *   <li>
     *     if <code>value</code> is not <code>null</code> and not
     *     an instance of type that is specified in by the given
     *     <code>key</code> (see {@link AttributeKey#getType()})
     *   </li>
     *  </ul>
     * 
     * @see #getAttribute(AttributeKey)
     */
    //@Override
    //public <T> T setAttribute(AttributeKey<? extends T> key, T value) {
    //    return attributes.setAttribute(key, value);
    //};

    /**
     * {@inheritDoc}
     * 
     * @see Collections#unmodifiableSet(Set)
     */
    //@Override
    //public Set<AttributeKey<?>> getAttributeKeys() {
    //    return attributes.getAttributeKeys();
    //}

    /**
     * {@inheritDoc}
     * 
     * @exception IllegalArgumentException
     *                if <code>key==null</code>
     */
    //@Override
    //public <T> T removeAttribute(AttributeKey<T> key) {
    //    return attributes.removeAttribute(key);
    //}

    /**
     * {@inheritDoc}
     */
    public void write(Object message) {
        doWriteWithFuture(message, null);
    }

    /**
     * {@inheritDoc}
     */
    //public IoFuture<Void> writeWithFuture(Object message) {
    //    IoFuture<Void> future = new DefaultWriteFuture();
    //    doWriteWithFuture(message, future);
    //    return future;
    //}

    private void doWriteWithFuture(Object message, IoFuture<bool> future) {
        //LOG.Debug("writing message {} to session {}", message, this);

        if ((state == SessionState.CLOSED) || (state == SessionState.CLOSING)) {
            LOG.Error("writing to closed or closing session, the message is discarded");
            return;
        }

        // process the queue
        processMessageWriting(message, future);
    }

    /**
     * {@inheritDoc}
     */
    public WriteRequest enqueueWriteRequest(Object message) {
        WriteRequest request = null;

        //try {
        //    // Lock the queue while the message is written into it
        //    writeQueueReadLock.lock();

        //    if (isConnectedSecured()) {
        //        // SSL/TLS : we have to encrypt the message
        //        SslHelper sslHelper = getAttribute(SSL_HELPER, null);

        //        if (sslHelper == null) {
        //            throw new IllegalStateException();
        //        }

        //        request = sslHelper.processWrite(this, message, writeQueue);
        //    } else {
        //        // Plain message
        //        request = new DefaultWriteRequest(message);

        //        writeQueue.add(request);
        //    }
        //} finally {
        //    writeQueueReadLock.unlock();
        //}

        //// If it wasn't, we register this session as interested to write.
        //// It's done in atomic fashion for avoiding two concurrent registering.
        //if (!registeredForWrite.getAndSet(true)) {
        //    writeProcessor.flush(this);
        //}

        return request;
    }

    public void setNotRegisteredForWrite() {
        //registeredForWrite.set(false);
    }

    /**
     * {@inheritDoc}
     */
    public Queue<WriteRequest> acquireWriteQueue() {
        //writeQueueWriteLock.lock();
        return writeQueue;
    }

    /**
     * {@inheritDoc}
     */
    public void releaseWriteQueue() {
        //writeQueueWriteLock.unlock();
    }

    //------------------------------------------------------------------------
    // Event processing using the filter chain
    //------------------------------------------------------------------------

    /**
     * process session create event using the filter chain. To be called by the session {@link SelectorProcessor} .
     */
    public void processSessionCreated() {
        //LOG.debug("processing session created event for session {}", this);

        foreach (IoFilter filter in chain) {
            filter.sessionCreated(this);
        }
    }

    /**
     * process session opened event using the filter chain. To be called by the session {@link SelectorProcessor} .
     */
    public void processSessionOpened() {
        LOG.Debug("processing session open event");

        foreach (IoFilter filter in chain) {
            filter.sessionOpened(this);
        }
    }

    /**
     * process session closed event using the filter chain. To be called by the session {@link SelectorProcessor} .
     */
    public void processSessionClosed() {
        LOG.Debug("processing session closed event");

        foreach (IoFilter filter in chain) {
            filter.sessionClosed(this);
        }
    }

    /**
     * process session idle event using the filter chain. To be called by the session {@link SelectorProcessor} .
     */
    public void processSessionIdle(IdleStatus status) {
        //LOG.Debug("processing session idle {} event for session {}", status, this);

        foreach (IoFilter filter in chain) {
            filter.sessionIdle(this, status);
        }
    }

    /**
     * process session message received event using the filter chain. To be called by the session {@link SelectorProcessor} .
     * @param message the received message 
     */
    public void processMessageReceived(StreamBuffer message) {
        //LOG.debug("processing message '{}' received event for session {}", message, this);

        // save basic statistics 
        readBytes += message.Remaining();
        lastReadTime = (long)(DateTime.Now - DateTime.Parse("1970-1-01 00:00:00.000")).TotalMilliseconds;

        if (chain.Length < 1) {
            LOG.Debug("Nothing to do, the chain is empty");
        } else {
            readChainPosition = 0;
            // we call the first filter, it's supposed to call the next ones using the filter chain controller
            chain[readChainPosition].messageReceived(this, message, this);
        }
    }

    /**
     * process session message writing event using the filter chain. To be called by the session {@link SelectorProcessor} .
     * @param message the wrote message, should be transformed into ByteBuffer at the end of the filter chain 
     */
    public void processMessageWriting(Object message, IoFuture<bool> future) {
        //LOG.debug("processing message '{}' writing event for session {}", message, this);

        lastWriteRequest = null;

        if (chain.Length < 1) {
            enqueueFinalWriteMessage(message);
        } else {
            writeChainPosition = chain.Length - 1;
            // we call the first filter, it's supposed to call the next ones using the filter chain controller
            int position = writeChainPosition;
            IoFilter nextFilter = chain[position];
            nextFilter.messageWriting(this, message, this);
        }

        // put the future in the last write request
        if (future != null) {
            WriteRequest request = lastWriteRequest;

            if (request != null) {
                //((DefaultWriteRequest) request).setFuture(future);
            }
        }
    }

    /**
     * process session message received event using the filter chain. To be called by the session {@link SelectorProcessor} .
     * @param message the received message 
     */
    public void callWriteNextFilter(Object message) {
        //LOG.debug("calling next filter for writing for message '{}' position : {}", message, writeChainPosition);

        writeChainPosition--;

        if (writeChainPosition < 0 || chain.Length == 0) {
            // end of chain processing
            enqueueFinalWriteMessage(message);
        } else {
            chain[writeChainPosition].messageWriting(this, message, this);
        }

        writeChainPosition++;
        ;
    }

    /**
     * At the end of write chain processing, enqueue encoded {@link ByteBuffer} message in the session
     */
    private void enqueueFinalWriteMessage(Object message) {
        //LOG.Debug("end of write chan we enqueue the message in the session : {}", message);
        lastWriteRequest = enqueueWriteRequest(message);
    }

    /**
     * {@inheritDoc}
     */
    public void callReadNextFilter(Object message) {
        readChainPosition++;

        if (readChainPosition >= chain.Length) {
            // end of chain processing
        } else {
            chain[readChainPosition].messageReceived(this, message, this);
        }

        readChainPosition--;
    }

    public string SessionID
    {
        get { throw new NotImplementedException(); }
    }

    public string IdentityKey
    {
        get { throw new NotImplementedException(); }
    }

    public System.Net.IPEndPoint RemoteAddress
    {
        get { throw new NotImplementedException(); }
    }

    public System.Net.IPEndPoint LocalAddress
    {
        get { throw new NotImplementedException(); }
    }

    public IoService GetService
    {
        get { throw new NotImplementedException(); }
    }

    bool IoSession.isClosed
    {
        get { throw new NotImplementedException(); }
    }

    bool IoSession.isClosing
    {
        get { throw new NotImplementedException(); }
    }

    bool IoSession.isConnected
    {
        get { throw new NotImplementedException(); }
    }

    bool IoSession.isCreated
    {
        get { throw new NotImplementedException(); }
    }

    bool IoSession.isSecuring
    {
        get { throw new NotImplementedException(); }
    }

    bool IoSession.isSecured
    {
        get { throw new NotImplementedException(); }
    }

    public void Close(bool immediately)
    {
        throw new NotImplementedException();
    }

    public void suspendRead()
    {
        throw new NotImplementedException();
    }

    public void suspendWrite()
    {
        throw new NotImplementedException();
    }

    public void resumeRead()
    {
        throw new NotImplementedException();
    }

    public void resumeWrite()
    {
        throw new NotImplementedException();
    }

    public bool isReadSuspended()
    {
        throw new NotImplementedException();
    }

    public bool isWriteSuspended()
    {
        throw new NotImplementedException();
    }

    public IoSessionConfig getConfig()
    {
        throw new NotImplementedException();
    }

    public IoFuture<bool> writeWithFuture(object message)
    {
        throw new NotImplementedException();
    }
    }
}